package com.shujia.core

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.text.SimpleDateFormat
import java.time.Duration
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo11CardFLow {
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(1)

    //1、读取数据

    val source: KafkaSource[String] = KafkaSource.builder[String]
      .setBootstrapServers("master:9092")
      .setTopics("cars")
      .setGroupId("my-group")
      .setStartingOffsets(OffsetsInitializer.latest()) //读取最新的数据
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build

    val carsDS: DataStream[String] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")


    //2、解析数据，取出卡口，时间戳，车速
    val catdDS: DataStream[(Long, Long, Double)] = carsDS.map(car => {
      val carJson: JSONObject = JSON.parseObject(car)

      val card: Long = carJson.getLong("card")
      val time: Long = carJson.getLong("time")
      val speed: Double = carJson.getDouble("speed")

      (card, time, speed)
    })


    //3、设置时间字段和水位线
    val assDS: DataStream[(Long, Long, Double)] = catdDS.assignTimestampsAndWatermarks(
      WatermarkStrategy
        //水位线前移5秒，最大允许数据乱序5秒，窗口延迟5秒计算
        .forBoundedOutOfOrderness[(Long, Long, Double)](Duration.ofSeconds(5))
        //指定时间字段
        .withTimestampAssigner(
        new SerializableTimestampAssigner[(Long, Long, Double)] {
          override def extractTimestamp(element: (Long, Long, Double), recordTimestamp: Long): Long = {
            //时间字段, flink需要毫秒级别的时间戳
            element._2 * 1000
          }
        })
    )

    //3、按照卡口分分组，划分窗口
    val windowDS: WindowedStream[(Long, Long, Double), Long, TimeWindow] = assDS
      .keyBy(_._1)
      .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))

    //4、卡口车流量，平均车速
    val cardAvgSpeedAndFlow: DataStream[(Long, Long, Double, String)] = windowDS.process(new CardWindowFunction)

    //3、将数据保存到mysql中
    cardAvgSpeedAndFlow.addSink(new CardMysqlSink)

    env.execute()

    /**
      * 1、生产数据
      * kafka-console-producer.sh --broker-list master:9092 --topic cars
      * {"com.shujia.batch.car":"皖A9A7N2","city_code":"340500","county_code":"340522","card":117988031603010,"camera_id":"00001","orientation":"西南","road_id":34052055,"time":1614711895,"speed":36.38}
      * {"com.shujia.batch.car":"皖A9A7N2","city_code":"340500","county_code":"340522","card":117988031603010,"camera_id":"01001","orientation":"西南","road_id":34052056,"time":1614711904,"speed":35.38}
      * {"com.shujia.batch.car":"皖A9A7N2","city_code":"340500","county_code":"340522","card":117988031603010,"camera_id":"01214","orientation":"西南","road_id":34052057,"time":1614711914,"speed":45.38}
      * {"com.shujia.batch.car":"皖A9A7N2","city_code":"340500","county_code":"340522","card":117988031603010,"camera_id":"01214","orientation":"西南","road_id":34052057,"time":1614711965,"speed":45.38}
      *
      */


  }
}

class CardWindowFunction extends ProcessWindowFunction[(Long, Long, Double), (Long, Long, Double, String), Long, TimeWindow] {
  /**
    * process: 一个key的一个窗口执行一次
    *
    * @param card     ： 卡口
    * @param context  ： 上下文对象，可以获取到窗口的开始和结束时间
    * @param elements ： 窗口的数据
    * @param out      ：用于将数据发送到下游
    */
  override def process(card: Long,
                       context: Context,
                       elements: Iterable[(Long, Long, Double)],
                       out: Collector[(Long, Long, Double, String)]): Unit = {
    //获取窗口的结束时间，用于标识，当前卡口什么时候拥堵
    val windowEnd: Long = context.window.getEnd
    //将时间戳转换成时间字符串
    val date = new Date(windowEnd)
    val format = new SimpleDateFormat("yyyyMMddHHmmss")
    val dateStr: String = format.format(date)


    var sumSpeed = 0.0
    var flow = 0

    for ((_, _, speed: Double) <- elements) {

      //统计总的车速
      sumSpeed += speed

      //统计总的车流量
      flow += 1
    }

    //计算平均车速
    val avgSpeed: Double = sumSpeed / flow

    //将数据发送到下游
    out.collect((card, flow, avgSpeed, dateStr))
  }
}


class CardMysqlSink extends RichSinkFunction[(Long, Long, Double, String)] {
  var con: Connection = _
  var stat: PreparedStatement = _

  override def invoke(value: (Long, Long, Double, String), context: SinkFunction.Context): Unit = {
    val (card, flow, avgSpeed, dateStr) = value
    stat.setLong(1, card)
    stat.setLong(2, flow)
    stat.setDouble(3, avgSpeed)
    stat.setString(4, dateStr)

    stat.execute()
  }

  override def open(parameters: Configuration): Unit = {
    //加载驱动
    Class.forName("com.mysql.jdbc.Driver")
    //创建链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
    stat = con.prepareStatement("insert into card_avg_speed_and_flow(card,flow,avgSpeed,dateStr) values(?,?,?,?)")


  }

  override def close(): Unit = {
    con.close()
    stat.close()
  }
}